use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
use blake2::Digest;
use bytes::Bytes;
use qp2p::{Config, Endpoint, RetryConfig};
use xor_name::XorName;
use serde::{Deserialize, Serialize};

use tracing::{debug, error, info, trace, warn};

use crate::metadata::Attribute;
use crate::remote::NodeRemote;
use crate::Scheduler;


#[derive(Debug, Clone, Deserialize, Serialize)]
pub enum Event {
    Ping {
        id: u64,
    },
    Pong {
        id: u64,
    },
    Metadata(Vec<(PathBuf, Attribute)>),
    Contacts(String),
    Create ((PathBuf, Attribute)), // when create, only create file in local
    Rename {
        source: PathBuf,
        target: PathBuf,
    },
    Read {
        id: u64,
        fh: u64,
        size: u32,
        offset: u64,
    },
    ReadResponse {
        id: u64,
        fh: u64,
        code: u32,
        data: Bytes,
    },
    Write {
        id: u64,
        fh: u64,
        data: Bytes,
        offset: u64,
    },
    WriteResponse {
        id: u64,
        fh: u64,
        code: u32,
        size: u32,
    },
    Open {
        id: u64,
        path: PathBuf,
    },
    Opened {
        id: u64,
        fh: u64,
        code: u32,
    },
    Remove(PathBuf),
    Rmdir(PathBuf),
    Release {
        id: u64,
        fh: u64,
    },
    Released {
        id: u64,
        fh: u64,
        code: u32,
    },
    Attribute{
        path: PathBuf,
        attr: Attribute,
    },
}

#[derive(Clone, Debug)]
pub enum Request {
    Metadata {
        data: Vec<(PathBuf, Attribute)>,
        peer: XorName,
    },
    Contacts {
        data: String,
        peer: XorName,
    },
    Create {
        data: (PathBuf, Attribute),
        peer: XorName,
    },
    Rename {
        source: PathBuf,
        target: PathBuf,
        peer: XorName,
    },
    Open {
        id: u64,
        path: PathBuf,
        peer: XorName,
    },
    Read {
        id: u64,
        fh: u64,
        size: u32,
        offset: u64,
        peer: XorName,
    },
    Write {
        id: u64,
        fh: u64,
        data: Bytes,
        offset: u64,
        peer: XorName,
    },
    Remove {
        path: PathBuf,
        peer: XorName,
    },
    Rmdir {
        path: PathBuf,
        peer: XorName,
    },
    Release {
        id: u64,
        fh: u64,
        peer: XorName,
    },
    Attribute{
        path: PathBuf,
        attr: Attribute,
        peer: XorName,
    },
}

pub fn address_to_name(address: &SocketAddr) -> XorName {
    let hash = address.to_string();
    let hash = blake2::Blake2b::digest(hash.as_bytes());
    let mut name = [0_u8; 32];
    name.copy_from_slice(&hash[..32]);
    XorName::from_content(&name)
}

pub async fn run(scheduler: Arc<Scheduler>, local: SocketAddr, contacts: &[SocketAddr]) {
    let mut retry = RetryConfig::default();
    retry.initial_retry_interval = Duration::from_millis(100);
    retry.max_retry_interval = Duration::from_millis(500);
    retry.retrying_max_elapsed_time = Duration::from_millis(500);
    let config = Config {
        idle_timeout: Duration::from_secs(60).into(), // 1 hour idle timeout.
        keep_alive_interval: Duration::from_secs(60).into(),
        retry_config: retry,
        ..Default::default()
    };
    let (endpoint, mut connections, contact) =
        Endpoint::new_peer(local, contacts, config).await.unwrap();
    let my_id = address_to_name(&endpoint.clone().public_addr());
    let (sender, mut receiver) = tokio::sync::mpsc::channel::<Request>(256);

    let remote_sender = sender.clone();
    let schedule = scheduler.clone();
    if let Some((conn, incoming)) = contact {
        let generator = schedule.generator();
        info!("connection peer: {}", conn.id());
        let id = address_to_name(&conn.remote_address());
        if id != my_id {
            let remote = NodeRemote::run(id, conn, incoming, remote_sender.clone(), generator.clone());
            schedule.add_remote(id, remote).await;
        }
    }

    let remote_sender = sender.clone();
    let schedule = scheduler.clone();
    tokio::spawn(async move {
        let generator = schedule.generator();
        while let Some((conn, incoming)) = connections.next().await {
            info!("connection incoming: {}", conn.id());
            let id = address_to_name(&conn.remote_address());
            if !schedule.have_remote(&id).await {
                let remote = NodeRemote::run(id, conn, incoming, remote_sender.clone(), generator.clone());
                schedule.add_remote(id, remote).await;
            } else {
                info!("peer {} is exists.", id);
            }
        }
    });

    let schedule = scheduler.clone();
    let endpoint = endpoint.clone();
    let remote_sender = sender.clone();
    // handle event and send response
    tokio::spawn(async move {
        while let Some(event) = receiver.recv().await {
            match event {
                Request::Metadata {data, peer} => {
                    warn!("receive metadata from {}, size {} failed", peer, data.len());
                    schedule.store(peer, data).await
                }
                Request::Contacts {data, peer} => {
                    if let Ok(addr) = data.parse::<SocketAddr>() {
                        debug!("found peer from {}: {}", peer, addr);
                        let id = address_to_name(&addr);
                        let generator = schedule.generator();
                        if !schedule.have_remote(&id).await {
                            if let Ok((conn, incoming)) = endpoint.connect_to(&addr).await {
                                let remote = NodeRemote::run(id, conn, incoming, remote_sender.clone(), generator.clone());
                                schedule.add_remote(id, remote).await;
                            }
                        }
                    }
                }
                Request::Create {data, peer} => {
                    if let Err(e) = schedule.store_one(peer, data).await {
                        warn!("store remote node for peer {} failed. {}", peer, e);
                    }
                }
                Request::Rename {source, target, peer} => {
                    if let Err(e) = schedule.move_remote(peer, &source, &target).await {
                        warn!("rename remote node for peer {} failed. {}", peer, e);
                    }
                }
                Request::Open {id, path, peer} => {
                    trace!("receive open event local {}, {}", id, path.display());
                    let e = match schedule.open_path(&path).await {
                        Ok(fh) => {
                            Event::Opened{id, fh, code: 0}
                        },
                        Err(e) => {
                            warn!("open path {} request {} handle error: {}", path.display(), id, e);
                            Event::Opened {id, fh: 0, code: 1}
                        }
                    };
                    trace!("open file: {}, event {:?}", id, e);
                    if let Err(e)= schedule.send_remote(peer, e).await {
                        warn!("send open response to {} error: {}, {}, {}", peer, id, path.display(), e);
                    } else {
                        trace!("send open response to {}", peer);
                    }
                }
                Request::Read {id, fh, size, offset, peer} => {
                    trace!("receive read event local {}, {}, {}, {}, {}", id, fh, size, offset, peer);
                    let e = match schedule.read(id, fh, offset, size).await {
                        Ok(data) => {
                            Event::ReadResponse{id, code: 0, data: Bytes::copy_from_slice(&data), fh}
                        }
                        Err(e) => {
                            warn!("read {} request {} handle error: {}", fh, id, e);
                            Event::ReadResponse {id, fh, code: 1, data: Bytes::new()}
                        }
                    };
                    if let Err(e)= schedule.send_remote(peer, e).await {
                        warn!("send read response to {} error: {}, {}, {}", peer, id, offset, e);
                    }
                }
                Request::Write {id, fh, offset, data, peer} => {
                    trace!("receive write event local {}, {}, {}, {}, {}", id, fh, data.len(), offset, peer);
                    let e = match schedule.write(id, fh, offset, data.as_ref()).await {
                        Ok(size) => {
                            Event::WriteResponse{id, fh, code: 0, size}
                        }
                        Err(e) => {
                            warn!("write {} request {} handle error: {}", fh, id, e);
                            Event::WriteResponse {id, fh, code: 1, size: 0}
                        }
                    };
                    if let Err(e)= schedule.send_remote(peer, e).await {
                        warn!("send write response to {} error: {}, {}, {}", peer, id, offset, e);
                    }
                }
                Request::Remove {path, peer} => {
                    if let Err(e) = schedule.remove_remote(peer, &path).await {
                        warn!("remove {} request handle error: {}", path.display(), e);
                    };
                }
                Request::Rmdir {path, peer} => {
                    if let Err(e) = schedule.rmdir_remote(peer, &path).await {
                        warn!("rmdir {} request handle error: {}", path.display(), e);
                    };
                }
                Request::Release {id, fh, peer} => {
                    let e = if let Err(e) = schedule.release(fh).await {
                        warn!("release {} request {} handle error: {}", fh, id, e);
                        Event::Released {id, fh, code: 1}
                    } else {
                        Event::Released {id, fh, code: 0}
                    };
                    if let Err(e)= schedule.send_remote(peer, e).await {
                        warn!("send release response to {} error: {}, {}, {}", peer, id, fh, e);
                    }
                }
                Request::Attribute {path, attr, peer: _peer} => {
                    if let Err(e) = schedule.update(&path, attr).await {
                        warn!("update node {} attribute error: {}", path.display(), e);
                    }
                }
            }
        }
    });
}